package com.run.fjy.mr;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.run.bcpimp.builder.IntegerValueBuilder;
import com.run.bcpimp.builder.LongValueBuilder;
import com.run.bcpimp.builder.StringValueBuilder;
import com.run.bcpimp.conf.BcpExtractorParser;
import com.run.bcpimp.conf.DbFieldConfig;
import com.run.bcpimp.conf.DbTableConfig;
import com.run.bcpimp.exception.BcpConfigurationException;
import com.run.bcpimp.exception.BcpParseException;
import com.run.bcpimp.mr.io.ExtractTableValueWritable;
import com.run.bcpimp.processor.ResourceFactory;
import com.run.bcpimp.util.BcpUtils;
import com.run.bcpimp.util.MRConst;
import com.run.fjy.util.DateUtil;

public class PerDayReducer extends Reducer<Text,ExtractTableValueWritable,Text,MapWritable> {
	
	private static Logger logger = LoggerFactory.getLogger(PerDayReducer.class);

	private Map<Character, List<DbFieldConfig>> dbFieldMeta = new HashMap<Character, List<DbFieldConfig>>();

	
	private Set<Integer> dates = new HashSet<Integer>();

	private Context context;

	private List<DbFieldConfig> dbFieldsList;

	private char tableCode;

	private String type;


	@Override
	protected void cleanup(Context context) throws IOException,
			InterruptedException {
		dates.clear();dates=null;
	}


	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		super.setup(context);
		this.context = context;
		Configuration configuration = context.getConfiguration();
		type = context.getConfiguration().get("statis.unit","day");
		ResourceFactory rf = new ResourceFactory();
		rf.setup(configuration);
		logger.info("setup resource-factory");
		
		BcpExtractorParser parser = new BcpExtractorParser(configuration.get(MRConst.CFGFILE_BCPEXTRACT_KEY));
		logger.info("init bcp-extrqact-parser");
		Map<String, DbTableConfig> dbTableConfigMap = parser.getDbTableConfigMap();
		for(Entry<String, DbTableConfig> entry : dbTableConfigMap.entrySet()) {
			//初始化每张表的字段的 valuebuileder，并使用code作为key缓存起来供后续的reduce逻辑使用
			dbFieldMeta.put(entry.getValue().code.charAt(0), entry.getValue().getDbFieldsList());
			
			for (DbFieldConfig dfConfig : entry.getValue().getDbFieldsList()) {
				if("string".equals(dfConfig.type)) {
					dfConfig.valueBuilder = new StringValueBuilder();
				}  else if("int".equals(dfConfig.type)) {
					if(10>dfConfig.len) {
						dfConfig.valueBuilder = new IntegerValueBuilder();
					} else {
						dfConfig.valueBuilder = new LongValueBuilder();
					}
				} else {
					throw new BcpConfigurationException("db field "+dfConfig.name+"'s valuetype["+dfConfig.type+"] not support");
				}
				dfConfig.valueBuilder.setup("name["+dfConfig.name+"] type["+dfConfig.type+"]", null);
			}
		}
		
		logger.info("init db field meta ok, codes"+dbFieldMeta.keySet());
	}



	protected void reduce(Text key, Iterable<ExtractTableValueWritable> value,Context context)
			throws IOException, InterruptedException {
		
		tableCode = (char)key.getBytes()[0];
		dbFieldsList = dbFieldMeta.get(tableCode);
		MapWritable map = new MapWritable();
		
		for (ExtractTableValueWritable valueOne : value) {
			
			if(!dates.add(DateUtil.getDate(type, valueOne.getLastTime()))) {
				continue;
			}
			
			Text colValues = valueOne.getColValues();
			map.put(new Text("md5code"), key);
			buildMap(map,colValues);
			
			context.write(key, map);
		}		

		try {
		} finally {
			dates.clear();
			map.clear();
		}
		

	}
	
	
	private void buildMap(MapWritable map, Text colValues) throws BcpParseException {
		
		
		byte[] bytes = colValues.getBytes();
		int length = colValues.getLength();
		
		int colIndex = 0;
		int pos = 0;
		int start = 0;
		int clen = 0;
		logger.info("colValues.toString["+colValues.toString()+"], colVaues.byts["+new String(bytes)+"], bytes.length["+bytes.length+"], text.getLength()["+colValues.getLength()+"]");
		
		for(; pos<length;pos++) {
			if(bytes[pos] == BcpUtils.EXTRACT_COLVALUES_SPLIT) {
				clen = pos - start;
				DbFieldConfig dbFieldConfig = dbFieldsList.get(colIndex);
				if(clen>dbFieldConfig.len) {
					context.getCounter(BcpUtils.MR_COUNTER_GROUP, "error: value len exceed --"+tableCode+"."+dbFieldConfig.name);
				} else if(clen>0) {
					Text fieldName = new Text();
					Text fieldValue = new Text();
					fieldName.set(dbFieldConfig.name);
					fieldValue.set(String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen)));
					logger.info("for=>"+fieldName+" =>"+fieldValue);
					map.put(fieldName, fieldValue);
				} 
				start = pos+1;
				colIndex++;
			}
		}
		if(colIndex>=dbFieldsList.size()) return ;
		clen = length-start;
		DbFieldConfig dbFieldConfig = dbFieldsList.get(colIndex);
		if(clen>0) {
			Text fieldName = new Text();
			Text fieldValue = new Text();
			fieldName.set(dbFieldConfig.name);

			fieldValue.set(String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen)));
			logger.info("if=>"+fieldName+" =>"+fieldValue);
			map.put(fieldName, fieldValue);
		
		} 
		
	}

}
